/**
 * FileName: SeckillConsumer
 *
 * @author: 阿张
 * Date:     2021/2/25 8:49
 */
package com.qx.mq;

import com.alibaba.fastjson.JSON;
import com.qx.pojo.SeckillCommodity;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@Component
public class SeckillConsumer {

    @Resource
    private RedisTemplate<String, Integer> redisTemplate;

    /**消费者组名*/
    @Value("${suning.rocketmq.consumerGroup}")
    private String consumerGroup;

    /**NameServer地址*/
    @Value("${suning.rocketmq.namesrvaddr}")
    private String namesrvAddr;

    private List<SeckillCommodity> list;

    @PostConstruct
    public void consumer(){
        System.err.println("init defaultMQPushConsumer");

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            consumer.subscribe("user-topic","user-tag");
            consumer.registerMessageListener((MessageListenerConcurrently)(list,context) ->{
                try {
                    for (MessageExt messageExt : list) {
                        String s = new String(messageExt.getBody());
                        List<SeckillCommodity> lists = JSON.parseArray(s, SeckillCommodity.class);
                        for (SeckillCommodity seckillCommodity : lists) {
                            redisTemplate.opsForValue().set("XSMS"+seckillCommodity.getCommodityId(),seckillCommodity.getQualifyQuantity());
                        }
                        //输出消息内容
                        System.out.println("消费消息:"+new String(messageExt.getBody()));
                    }
                }catch (Exception e){
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}

